package com.zhang.hadoop.flink.test2;

import com.zhang.hadoop.flink.base.Event;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

/**
 * @author: zhang yufei
 * @createTime:2022/5/8 11:03
 * @description:
 */
public class SourceTest {

    public static void main(String[] args) throws Exception {
        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //1.从文件中读取数据
        DataStreamSource<String> stream1 = env.readTextFile("flink/input/clicks.txt");
        //2.从集合中读取数据
        List<Integer> nums = new ArrayList<>();
        nums.add(2);
        nums.add(5);
        DataStreamSource<Integer> numStream = env.fromCollection(nums);

        List<Event> events = new ArrayList<>();
        events.add(new Event("yanghui", "./home", 1000L));
        events.add(new Event("yuping", "./cart", 2000L));
        DataStreamSource<Event> stream2 = env.fromCollection(events);

        DataStreamSource<Event> stream3 = env.fromElements(
                new Event("yangdan", "./yindao", 1000L),
                new Event("jingru", "./gangmen", 1000L)
        );

        //4.从socket文本流中读取
        DataStreamSource<String> stream4 = env.socketTextStream("localhost", 8000);

        stream1.print("1");
        numStream.print("nums");
        stream2.print("2");
        stream3.print("3");
        stream4.print("4");

        //5.从kafka中读取数据
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers","127.0.0.1:9092" );
        properties.setProperty("group.id","consumer-group" );
        properties.setProperty("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer" );
        properties.setProperty("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer" );
        properties.setProperty("auto.offset.reset","latest" );
        DataStreamSource<String> kafkaStream = env.addSource(new FlinkKafkaConsumer<String>("clicks", new SimpleStringSchema(), properties));
        kafkaStream.print();

        env.execute();
    }
}
